package com.shujia.flink.core

import org.apache.flink.api.common.eventtime.WatermarkStrategy
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.connector.kafka.source.KafkaSource
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow

object Demo5Parallelism {
  def main(args: Array[String]): Unit = {
    /**
     * 并行度
     * 1、默认并行度时1
     * 2、可以在代码中同意设置并行的, env.setParallelism(2)
     * 3、每一个算子可以单独设置并行度   优先级最高
     * 4、在提交任务时设置并行的  -p 2    优先级最低
     *
     * 在flink开发过程中一般不在代码中设置并行度，而是在提交任务时设置，好处是修改并行度不需要改代码，不需要重新打包
     *
     * task合并原则
     * 1、当并行度一样，同时没有产生shuffle会将上下游合并成一个task
     *
     * 在flink中设置并行度原则
     * 看数据量
     * tps：每秒多少条
     *
     * 如果是聚合类的计算每一个并行度的tps可以达到3000-5000条数据每秒
     * 如果是非聚合计算每一个并行度的tps可以达到10000-20000条数据每秒
     *
     * 比如现在数据源每秒有10万条数据，代码中有聚合计算，那么需要指定20-30个并行度
     *
     *
     */
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    //env.setParallelism(2)

    //创建一个多分区的topic
    //kafka-topics.sh --create --zookeeper master:2181,node1:2181,node2:2181/kafka --replication-factor 2 --partitions 2 --topic words
    val source: KafkaSource[String] = KafkaSource
      .builder[String]
      .setBootstrapServers("master:9092,node1:9092,node2:9092") //kafka集群列表
      .setTopics("words") //topic
      .setGroupId("Demo5Parallelism") //消费者组
      .setStartingOffsets(OffsetsInitializer.earliest()) //读取数据的位置
      .setValueOnlyDeserializer(new SimpleStringSchema()) //反序列化类
      .build

    //构建DataStream
    val kafkaSource: DataStream[String] = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source")

    val kvDS: DataStream[(String, Int)] = kafkaSource.map((_, 1))

    val keyByDS: KeyedStream[(String, Int), String] = kvDS.keyBy(_._1)

    val windowDS: WindowedStream[(String, Int), String, TimeWindow] = keyByDS
      .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))

    val countDS: DataStream[(String, Int)] = windowDS.sum(1)

    //单独设置并行度
    countDS.print()//.setParallelism(1)

    env.execute()


  }

}
